/*
 * QueueThread Pool Class
 * Copyright (C) Rick <adou1127@hotmail.com>
 *

 */


#ifndef __QUEUETHREADPOOL_H
#define __QUEUETHREADPOOL_H

#include "Common.h"
#include "Singleton.h"
#include "Mutex.h"
#include "Condition.h"

#include "ThreadStarter.h"
#include "NGLog.h"
#include "Threading/ThreadPool.h"
#include <deque>




//template< class T> class SERVER_DECL CQueueThreadPool :public ThreadBase,public Singleton<CQueueThreadPool<T> >
template< class T> class SERVER_DECL CQueueThreadPool :public ThreadBase
{
	uint32	m_min_thread;
	uint32  m_max_thread;
	
	uint32  m_active_thread;
	uint32  m_idle_thread;
	uint32  m_timeout;
	uint32  m_num_thread;

	Mutex  m_mutex;
	Condition  * m_cont;
	

	
	std::deque<T *> m_queue;
	ThreadRunableBase * m_runable;
	bool	m_poolRunning;

public:
	WGSERVER_INLINE CQueueThreadPool()
	{
		m_cont = new Condition(&m_mutex);
        	m_min_thread = 5;
        	m_max_thread = 20;
        	m_active_thread = 0;
        	m_idle_thread = 0;
		m_num_thread = 0;
        	m_timeout = 0;
        	m_runable = 0;
		m_poolRunning = true;
		
	}
	virtual WGSERVER_INLINE ~CQueueThreadPool()
	{
		 delete m_cont;
	}
	// call every 2 minutes or so.

	// call at startup
	WGSERVER_INLINE void Startup(ThreadRunableBase * runable, int min=5,int max=20,int timeout = 0, \
			 const char * queueName="QueueThreadPool")
	{

		m_runable = runable;
		m_min_thread = min;
		m_max_thread = max;
		m_active_thread = 0;
		m_idle_thread = 0;
		m_timeout = timeout;
		m_threadName = queueName;

		 m_cont->BeginSynchronized();
		 
		 for (uint32 i = 0 ;i < m_min_thread;++i)
		 {
			m_num_thread ++;
			ThreadPool.ExecuteTask(this);
		 }
		 m_cont->EndSynchronized();
	}


	//virtual from ThreadBase	
	WGSERVER_INLINE bool run()
	{
		while(m_poolRunning)
		{
			m_cont->BeginSynchronized();
			T * t=0;
			++m_idle_thread;
			if(!m_queue.size())
			{
			    if(m_min_thread < m_num_thread && m_idle_thread >= m_num_thread)
			    {
				--m_idle_thread;
				m_cont->EndSynchronized();
				//Log.Debug("QueueThreadPool","Thread : %s exiting.",m_threadName.c_str());
				break;
			    }
		     	    
			   if(m_timeout)
			   {
			        if(!m_cont->Wait(m_timeout))
				{
				    --m_idle_thread;
			            m_cont->EndSynchronized();
				    continue;	
				}
			   }
			   else
			   {
			      	m_cont->Wait();
			   }

			}
		
			--m_idle_thread;

			if(m_queue.size())
			{	
		 		t= m_queue.front();
				m_queue.pop_front();


				if(m_idle_thread ==0 && m_num_thread < m_max_thread)
				{
					if(m_queue.size() > m_min_thread)
					{
						ThreadPool.ExecuteTask(this);
						++m_num_thread;
						if(m_queue.size() > m_max_thread)
						{
							m_min_thread ++;
						}
					}
				}

			}
			else
			{
				t = 0;
			}


			
			
			++m_active_thread;
			m_cont->EndSynchronized();

			if(t && m_runable)
			{
			    if(m_runable->run(t))
			    {
				delete t;
			    }

			}

			m_cont->BeginSynchronized();
			--m_active_thread;
			m_cont->EndSynchronized();
		}

		
		m_cont->BeginSynchronized();
		--m_num_thread;
		if(m_num_thread ==0 )
		{
		       /* signal that all threads are finished */
                       m_cont->Broadcast();
		}
		m_cont->EndSynchronized();

        	return false;
	}
	// shutdown all threads
	WGSERVER_INLINE void Shutdown()
	{
	    	m_poolRunning = false;
		
	    /* wait from threads to exit */
		m_cont->BeginSynchronized();
		if(m_num_thread>0)
		{
		    	m_cont->Broadcast();
		}
		m_cont->EndSynchronized();
			

		m_cont->BeginSynchronized();
		while(m_num_thread >0)
		{
		    m_cont->Wait();
		}

		m_cont->EndSynchronized();

	}

	// prints some neat debug stats
	WGSERVER_INLINE void ShowStats()
	{
		m_cont->BeginSynchronized();
		Log.Debug("QueueThreadPool", "============ QueueThreadPool(%s) Status =============",m_threadName.c_str());
		Log.Debug("QueueThreadPool", "Min threads: %d,Max threads %d,current threads count %d", \
					 m_min_thread ,m_max_thread,m_num_thread);
		Log.Debug("QueueThreadPool", "Active Threads: %u", m_active_thread);
		Log.Debug("QueueThreadPool", "Idle Threads: %u",m_idle_thread );
		Log.Debug("QueueThreadPool", "Queue Requested: %u",m_queue.size());
		Log.Debug("QueueThreadPool", "================================================================");
		m_cont->EndSynchronized();
	}

	
	WGSERVER_INLINE bool  addQueue(T * element)
	{
	        m_cont->BeginSynchronized();
		m_queue.push_back(element);

		 /*release the signal from wakeup waiting thread */
		m_cont->Signal();
		m_cont->EndSynchronized();

		return true;
	}

};


#endif
